package com.example.mq.emqx;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * @author Jason
 * @PACKAGE_NAME com.example.emqx
 * @Description mqtt订阅端
 * @date 2023-11-18 14:33
 */
public class Subscriber {
    public static void main(String[] args) {
        String topic = "test/topic"; // MQTT 主题 -> 定义要订阅的主题，消息将在这个主题下发布
        String broker = "tcp://localhost:1883"; // MQTT 代理地址 -> 指定 MQTT 代理服务器的地址和端口
        String clientId1 = "JavaSubscriber1"; // 第一个客户端 ID -> 为第一个 MQTT 客户端实例指定一个唯一的标识符
        String clientId2 = "JavaSubscriber2"; // 第二个客户端 ID -> 为第二个 MQTT 客户端实例指定一个唯一的标识符
        MemoryPersistence persistence = new MemoryPersistence(); // 持久化存储，这里使用内存持久化

        try {
            // 创建第一个 MQTT 客户端
            MqttClient client1 = new MqttClient(broker, clientId1, persistence);
            connectAndSubscribe(client1, broker, topic);

            // 创建第二个 MQTT 客户端
            MqttClient client2 = new MqttClient(broker, clientId2, persistence);
            connectAndSubscribe(client2, broker, topic);

            // 保持应用运行以接收消息
            System.out.println("按 <Enter> 键退出");
            try {
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            }

            // 断开两个客户端的连接
            client1.disconnect();
            client2.disconnect();
            System.out.println("已断开连接");

        } catch (MqttException me) {
            me.printStackTrace();
        }
    }

    private static void connectAndSubscribe(MqttClient client, String broker, String topic) throws MqttException {
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(true); // 连接选项，设置清除会话 -> 确保每次连接都是一个新会话
        System.out.println("正在连接代理: " + broker + ", 客户端ID: " + client.getClientId());
        client.connect(connOpts);
        System.out.println("已连接");

        client.setCallback(new MqttCallback() {
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("接收到消息 [" + client.getClientId() + "]: " + topic + " - " + new String(message.getPayload())); // 消息到达时的操作
            }

            public void connectionLost(Throwable cause) {
                System.out.println("连接丢失 [" + client.getClientId() + "]!"); // 连接丢失时的操作
            }

            public void deliveryComplete(IMqttDeliveryToken token) {
                // 消息传递完成时的操作（通常订阅端不需要）
            }
        });

        client.subscribe(topic); // 订阅主题 -> 订阅定义的主题以接收消息
        System.out.println("已订阅主题: " + topic + ", 客户端ID: " + client.getClientId());
    }
}
